package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.util.HBaseUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;

/**
 * @author gufg
 * @since 2025-07-04 15:47
 */
public class DwsHBaseSinkFunc implements SinkFunction<String> {
    @Override
    public void invoke(String jsonStr, Context context) throws Exception {

        /*
         * {
         *   "driverId": "11",
         *   "driverName": "邵5bfb8fd1c7af69efee217c72fabec38b",
         *   "lat": "30.70332",
         *   "lng": "104.05368",
         *   "orderId": "17592719043682",
         *   "orderIdDesensitization": "175********682",
         *   "time": "1475976805"
         * }
         */

        /*  数据倾斜解决==字符串拼接
            rowkey = driverId + orderId
                     00021117592719043682
            hbase:
                0001|    0002|
                0002|    0003|
                0003|    0004|

           Rowkey设计原则：
                尽量简短
         */

        JSONObject jsonObj = JSONObject.parseObject(jsonStr);


        Put put = new Put(Bytes.toBytes(jsonObj.getString("orderId")));

        for (Map.Entry<String, Object> entry : jsonObj.entrySet()) {
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(entry.getKey()),Bytes.toBytes(String.valueOf(entry.getValue())));
        }

        HBaseUtil.put("dws_locus", put);

    }
}
